package com.vhall.component.plugin.core.service.init;

import io.lettuce.core.RedisClient;
import io.lettuce.core.RedisURI;
import io.lettuce.core.XGroupCreateArgs;
import io.lettuce.core.XReadArgs;
import io.lettuce.core.api.StatefulRedisConnection;
import io.lettuce.core.api.sync.RedisCommands;
import io.lettuce.core.cluster.RedisClusterClient;
import io.lettuce.core.cluster.api.StatefulRedisClusterConnection;
import io.lettuce.core.cluster.api.sync.RedisAdvancedClusterCommands;
import io.lettuce.core.cluster.api.sync.RedisClusterCommands;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.time.Duration;
import java.time.temporal.ChronoUnit;
import java.util.ArrayList;
import java.util.List;

/**
 * @author jingwen.li
 * @date 2022/2/8
 */
@Slf4j
@Component
public class RedisStreamInitialize {

    @Value("${spring.redis.model}")
    private String model;

    @Value("${spring.redis.host}")
    private String host;
    @Value("${spring.redis.port}")
    private Integer port;

    @Value("${spring.redis.cluster.nodes:}")
    private String nodes;

    @Value("${spring.redis.database}")
    private Integer database;
    @Value("${spring.redis.password}")
    private String password;

    @Value("${vhall.exam.stream-name:exam-stream}")
    public String examStreamName;
    @Value("${vhall.exam.group-name:exam-group}")
    private String examGroupName;


    @Value("${vhall.bigdata.broadcast-record-name:bigdata-broadcast-record-stream}")
    private String broadcastRecordReportStreamName;

    @Value("${vhall.bigdata.broadcast-record-group-name:bigdata-broadcast-record-group}")
    private String broadcastRecordReportGroupName;

    @Value("${vhall.export.admin-stream-name:export-admin-stream}")
    private String adminStreamName;
    @Value("${vhall.export.admin-group-name:export-admin-group}")
    private String adminGroupName;

    @Value("${vhall.red-packet.stream-name:rp-stream}")
    public String rpStreamName;
    @Value("${vhall.red-packet.group-name:rp-group}")
    private String rpGroupName;

    @Value("${vhall.lottery.stream-name:lottery-stream}")
    public String lotteryStreamName;

    @Value("${vhall.lottery.group-name:lottery-group}")
    private String lotteryGroupName;

    @Value("${vhall.export.stream-name:export-stream}")
    private String exportStreamName;
    @Value("${vhall.export.group-name:export-group}")
    private String exportGroupName;

    @PostConstruct
    public String streamAdd() {
        if ("single".equals(model)) {
            return singleInit();
        } else {
            return clusterInit();
        }
    }

    private String clusterInit() {
        RedisClusterClient redisClusterClient = null;
        try {
            log.warn("========> redis cluster stream 初始化 start");

            List<RedisURI> nodeList = new ArrayList<>();
            String[] nodeArray = nodes.split(",");
            for (String node : nodeArray) {
                String[] hostAndPort = node.split(":");
                String host = hostAndPort[0];
                String port = hostAndPort[1];
                RedisURI redisUri = RedisURI.builder()
                        .withHost(host)
                        .withPort(Integer.parseInt(port))
                        .withTimeout(Duration.of(10, ChronoUnit.SECONDS))
                        .withDatabase(database)
                        .withPassword(password)
                        .build();
                nodeList.add(redisUri);
            }
            redisClusterClient = RedisClusterClient.create(nodeList);
            StatefulRedisClusterConnection<String, String> connect = redisClusterClient.connect();
            RedisAdvancedClusterCommands<String, String> sync = connect.sync();
            XGroupCreateArgs args = new XGroupCreateArgs().mkstream(true);

            exportStreamInit(sync, args);
            examStreamInit(sync, args);
            broadcastRecordReportStreamInit(sync, args);
            adminStreamInit(sync, args);
            rpStreamInit(sync, args);
            lotteryStreamInit(sync, args);

            return "success";
        } catch (Exception ex) {
            log.error("========> redis cluster stream 初始化 异常", ex);
            throw ex;
        } finally {
            if (null != redisClusterClient) {
                redisClusterClient.connect().close();
            }
        }
    }

    private String singleInit() {
        RedisClient redisClient = null;
        try {
            log.warn("========> single redis stream 初始化 start");
            RedisURI redisUri = RedisURI.builder()
                    .withHost(host)
                    .withPort(port)
                    .withTimeout(Duration.of(10, ChronoUnit.SECONDS))
                    .withDatabase(database)
                    .withPassword(password)
                    .build();
            redisClient = RedisClient.create(redisUri);

            StatefulRedisConnection<String, String> connect = redisClient.connect();
            RedisCommands<String, String> redisSyncCommands = connect.sync();
            XGroupCreateArgs args = new XGroupCreateArgs().mkstream(true);

            exportStreamInit(redisSyncCommands, args);
            examStreamInit(redisSyncCommands, args);
            broadcastRecordReportStreamInit(redisSyncCommands, args);
            adminStreamInit(redisSyncCommands, args);
            rpStreamInit(redisSyncCommands, args);
            lotteryStreamInit(redisSyncCommands, args);

            return "success";
        } catch (Exception ex) {
            log.error("========> single redis stream 初始化 异常", ex);
            throw ex;
        } finally {
            if (null != redisClient) {
                redisClient.connect().close();
            }
        }
    }

    private void examStreamInit(RedisClusterCommands<String, String> redisSyncCommands, XGroupCreateArgs args) {
        try {
            Long size = redisSyncCommands.exists(examStreamName);
            if (size > 0) {
                log.warn("========> exam stream 初始化 ends,已存在,跳过 stream:[{}], group:[{}]", examStreamName, examGroupName);
                return;
            }
            // 创建分组
            String result = redisSyncCommands.xgroupCreate(XReadArgs.StreamOffset.from(examStreamName, "0"), examGroupName, args);

            //XGROUP CREATE mystream group-1 $ MKSTREAM
            log.warn("========> exam stream 初始化 ends, stream:[{}], group:[{}], result:[{}]", examStreamName, examGroupName, result);
        } catch (Exception e) {
            log.error("========> redis stream 初始化 异常", e);
            throw e;
        }
    }

    private void broadcastRecordReportStreamInit(RedisClusterCommands<String, String> redisSyncCommands, XGroupCreateArgs args) {
        try {
            Long size = redisSyncCommands.exists(broadcastRecordReportStreamName);
            if (size > 0) {
                log.warn("========> Bigdata Broadcast Record stream 初始化 ends,已存在,跳过 stream:[{}], group:[{}]", broadcastRecordReportStreamName, broadcastRecordReportGroupName);
                return;
            }
            // 创建分组
            String result = redisSyncCommands.xgroupCreate(XReadArgs.StreamOffset.from(broadcastRecordReportStreamName, "0"), broadcastRecordReportGroupName, args);

            log.warn("========> Bigdata Broadcast Record stream 初始化 ends, stream:[{}], group:[{}], result:[{}]", broadcastRecordReportStreamName, broadcastRecordReportGroupName, result);
        } catch (Exception e) {
            log.error("========> redis stream 初始化 异常", e);
            throw e;
        }
    }

    private void adminStreamInit(RedisClusterCommands<String, String> redisSyncCommands, XGroupCreateArgs args) {
        try {
            Long size = redisSyncCommands.exists(adminStreamName);
            if (size > 0) {
                log.warn("========> admin stream 初始化 ends,已存在,跳过 stream:[{}], group:[{}]", adminStreamName, adminGroupName);
                return;
            }
            // 创建分组
            String result = redisSyncCommands.xgroupCreate(XReadArgs.StreamOffset.from(adminStreamName, "0"), adminGroupName, args);

            log.warn("========> admin stream 初始化 ends, stream:[{}], group:[{}], result:[{}]", adminStreamName, adminGroupName, result);
        } catch (Exception e) {
            log.error("========> admin stream 初始化 异常", e);
            throw e;
        }
    }

    private void rpStreamInit(RedisClusterCommands<String, String> redisSyncCommands, XGroupCreateArgs args) {
        try {
            Long size = redisSyncCommands.exists(rpStreamName);
            if (size > 0) {
                log.warn("========> red packet stream 初始化 ends,已存在,跳过 stream:[{}], group:[{}]", rpStreamName, rpGroupName);
                return;
            }
            // 创建分组
            String result = redisSyncCommands.xgroupCreate(XReadArgs.StreamOffset.from(rpStreamName, "0"), rpGroupName, args);

            log.warn("========> red packet stream 初始化 ends, stream:[{}], group:[{}], result:[{}]", rpStreamName, rpGroupName, result);
        } catch (Exception e) {
            log.error("========> red packet stream 初始化 异常", e);
            throw e;
        }
    }

    private void lotteryStreamInit(RedisClusterCommands<String, String> redisSyncCommands, XGroupCreateArgs args) {
        try {
            Long size = redisSyncCommands.exists(lotteryStreamName);
            if (size > 0) {
                log.warn("========> lottery stream 初始化 ends,已存在,跳过 stream:[{}], group:[{}]", lotteryStreamName, lotteryGroupName);
                return;
            }
            // 创建分组
            String result = redisSyncCommands.xgroupCreate(XReadArgs.StreamOffset.from(lotteryStreamName, "0"), lotteryGroupName, args);

            log.warn("========> lottery stream 初始化 ends, stream:[{}], group:[{}], result:[{}]", lotteryStreamName, lotteryGroupName, result);
        } catch (Exception e) {
            log.error("========> lottery stream 初始化 异常", e);
            throw e;
        }
    }


    private void exportStreamInit(RedisClusterCommands<String, String> redisSyncCommands, XGroupCreateArgs args) {
        try {
            Long size = redisSyncCommands.exists(exportStreamName);
            if (size > 0) {
                log.warn("========> export stream 初始化 ends,已存在,跳过 stream:[{}], group:[{}]", exportStreamName, exportGroupName);
                return;
            }
            // 创建分组
            String result = redisSyncCommands.xgroupCreate(XReadArgs.StreamOffset.from(exportStreamName, "0"), exportGroupName, args);

            log.warn("========> export stream 初始化 ends, stream:[{}], group:[{}], result:[{}]", exportStreamName, exportGroupName, result);
        } catch (Exception e) {
            log.error("========> export stream 初始化 异常", e);
            throw e;
        }
    }
}
